package drds.global_transaction.resource_manager;

import drds.common.Constants;
import drds.common.ShouldNeverHappenException;
import drds.global_transaction.api.api.BranchTransactionStatus;
import drds.global_transaction.api.api.IResourceManager;
import drds.global_transaction.api.api.Resource;
import drds.global_transaction.api.api.ResourceManagerInbound;
import drds.global_transaction.api.exception.GlobalTransactionException;
import drds.global_transaction.api.exception.TransactionExceptionCode;
import drds.global_transaction.api.message_codec.ResponseCode;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.branch_transaction_register.BranchTransactionRegisterRequestMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.branch_transaction_register.BranchTransactionRegisterResponseMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.branch_transaction_status_report.BranchTransactionStatusReportRequestMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.branch_transaction_status_report.BranchTransactionStatusReportResponseMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.global_transaction_lock_record_ids_check.GlobalTransactionLockRecordIdsCheckRequestMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.global_transaction_lock_record_ids_check.GlobalTransactionLockRecordIdsCheckResponseMessage;
import drds.global_transaction.api.rpc.rpc_client.resource_manager.ResourceManagerRpcClient;
import drds.global_transaction.common.GlobalTransactionIds;
import drds.global_transaction.resource_manager.proxy.resource_manager.DataSourceProxy;
import drds.global_transaction.resource_manager.undolog.UndoLogManager;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeoutException;

@Slf4j
public class ResourceManager implements IResourceManager
{

	private static class SingletonHolder
	{
		private static ResourceManager INSTANCE = new ResourceManager();
	}

	//
	//
	public static ResourceManager get()
	{
		return SingletonHolder.INSTANCE;
	}

	public static synchronized void init(ResourceManagerInbound asyncWorker)
	{
		get().setAsyncWorker(asyncWorker);
	}

	public static void set(ResourceManager mock)
	{
		SingletonHolder.INSTANCE = mock;
	}

	private ResourceManagerInbound asyncWorker;

	private Map<String, Resource> resourceIdToResourceMap = new ConcurrentHashMap<>();

	protected ResourceManager()
	{
	}

	@Override
	public BranchTransactionStatus branchTransactionCommit(String globalTransactionIdWithAddressAndPort, long branchTransactionId, String resourceId, String applicationData) throws GlobalTransactionException
	{
		return asyncWorker.branchTransactionCommit(globalTransactionIdWithAddressAndPort, branchTransactionId, resourceId, applicationData);
	}

	//
	@Override
	public Long branchTransactionRegister(String resourceId, String clientIp, String globalTransactionIdWithAddressAndPort, String tableNameAndLockRecordIdsStringsString) throws GlobalTransactionException
	{
		try
		{
			BranchTransactionRegisterRequestMessage branchTransactionRegisterRequestMessage = new BranchTransactionRegisterRequestMessage();
			branchTransactionRegisterRequestMessage.setGlobalTransactionId(GlobalTransactionIds.getGlobalTransactionId(globalTransactionIdWithAddressAndPort));
			branchTransactionRegisterRequestMessage.setResourceId(resourceId);
			branchTransactionRegisterRequestMessage.setTableNameAndLockRecordIdsStringsString(tableNameAndLockRecordIdsStringsString);
			//
			BranchTransactionRegisterResponseMessage branchTransactionRegisterResponseMessage = //
					(BranchTransactionRegisterResponseMessage) ResourceManagerRpcClient.getInstance().asynchronousSendRequestWithTimeoutGetResponse(//
							branchTransactionRegisterRequestMessage);//
			if (branchTransactionRegisterResponseMessage.getResponseCode() == ResponseCode.failed)
			{
				log.error("分支事务注册失败:" + branchTransactionRegisterResponseMessage.getTransactionExceptionCode());
				throw new GlobalTransactionException(branchTransactionRegisterResponseMessage.getTransactionExceptionCode(), "Response[" + branchTransactionRegisterResponseMessage.getMessage() + "]");
			}
			return branchTransactionRegisterResponseMessage.getBranchTransactionId();
		} catch (TimeoutException toe)
		{
			throw new GlobalTransactionException(TransactionExceptionCode.io, "RPC Timeout", toe);
		} catch (RuntimeException rex)
		{
			if (Constants.developMode)
			{
				rex.printStackTrace();
			}
			log.error(rex.getMessage());
			throw new GlobalTransactionException(TransactionExceptionCode.branch_transaction_register_failed, "Runtime", rex);
		}
	}

	@Override
	public void branchTransactionReport(String globalTransactionIdWithAddressAndPort, long branchTransactionId, BranchTransactionStatus branchTransactionStatus, String applicationData) throws GlobalTransactionException
	{
		try
		{
			BranchTransactionStatusReportRequestMessage branchTransactionStatusReportRequestMessage = new BranchTransactionStatusReportRequestMessage();
			branchTransactionStatusReportRequestMessage.setGlobaltransactionId(GlobalTransactionIds.getGlobalTransactionId(globalTransactionIdWithAddressAndPort));
			branchTransactionStatusReportRequestMessage.setBranchTransactionId(branchTransactionId);
			branchTransactionStatusReportRequestMessage.setBranchTransactionStatus(branchTransactionStatus);
			//
			branchTransactionStatusReportRequestMessage.setApplicationData(applicationData);

			BranchTransactionStatusReportResponseMessage branchTransactionStatusReportResponseMessage = (BranchTransactionStatusReportResponseMessage) ResourceManagerRpcClient.getInstance().//
					asynchronousSendRequestWithTimeoutGetResponse(branchTransactionStatusReportRequestMessage);//
			if (branchTransactionStatusReportResponseMessage.getResponseCode() == ResponseCode.failed)
			{
				throw new GlobalTransactionException(branchTransactionStatusReportResponseMessage.getTransactionExceptionCode(), "Response[" + branchTransactionStatusReportResponseMessage.getMessage() + "]");
			}
		} catch (TimeoutException toe)
		{
			throw new GlobalTransactionException(TransactionExceptionCode.io, "RPC Timeout", toe);
		} catch (RuntimeException rex)
		{
			if (Constants.developMode)
			{
				rex.printStackTrace();
			}
			log.error(rex.getMessage());
			throw new GlobalTransactionException(TransactionExceptionCode.branch_transaction_report_failed, "Runtime", rex);
		}

	}

	@Override
	public BranchTransactionStatus branchTransactionRollback(String globalTransactionIdWithAddressAndPort, long branchTransactionId, String resourceId, String applicationData) throws GlobalTransactionException
	{
		DataSourceProxy dataSourceProxy = get(resourceId);
		if (dataSourceProxy == null)
		{
			throw new ShouldNeverHappenException();
		}
		try
		{
			UndoLogManager.undo(dataSourceProxy, globalTransactionIdWithAddressAndPort, branchTransactionId);
		} catch (GlobalTransactionException e)
		{
			// 回滚失败状态
			if (e.getTransactionExceptionCode() == TransactionExceptionCode.branch_transaction_rollback_failed_and_unretriable)
			{
				return BranchTransactionStatus.phase_two_rollback_failed_and_unretriable;
			} else
			{
				return BranchTransactionStatus.phase_two_rollback_failed_and_retriable;
			}
		}
		return BranchTransactionStatus.phase_two_rollbacked;

	}

	public DataSourceProxy get(String resourceId)
	{
		return (DataSourceProxy) resourceIdToResourceMap.get(resourceId);
	}

	@Override
	public Map<String, Resource> getResourceIdToResourceMap()
	{
		return resourceIdToResourceMap;
	}

	@Override
	public boolean lockRecordIdsCheck(String resourceId, String globalTransactionIdWithAddressAndPort, String tableNameAndLockRecordIdsStringsString) throws GlobalTransactionException
	{
		try
		{
			GlobalTransactionLockRecordIdsCheckRequestMessage globalTransactionLockRecordIdsCheckRequestMessage = new GlobalTransactionLockRecordIdsCheckRequestMessage();
			globalTransactionLockRecordIdsCheckRequestMessage.setGlobalTransactionId(GlobalTransactionIds.getGlobalTransactionId(globalTransactionIdWithAddressAndPort));
			globalTransactionLockRecordIdsCheckRequestMessage.setResourceId(resourceId);
			globalTransactionLockRecordIdsCheckRequestMessage.setTableNameAndLockRecordIdsStringsString(tableNameAndLockRecordIdsStringsString);

			GlobalTransactionLockRecordIdsCheckResponseMessage globalTransactionLockRecordIdsCheckResponseMessage = (GlobalTransactionLockRecordIdsCheckResponseMessage) ResourceManagerRpcClient.getInstance().//
					asynchronousSendRequestWithTimeoutGetResponse(globalTransactionLockRecordIdsCheckRequestMessage);
			if (globalTransactionLockRecordIdsCheckResponseMessage.getResponseCode() == ResponseCode.failed)
			{
				throw new GlobalTransactionException(globalTransactionLockRecordIdsCheckResponseMessage.getTransactionExceptionCode(), "Response[" + globalTransactionLockRecordIdsCheckResponseMessage.getMessage() + "]");
			}
			return globalTransactionLockRecordIdsCheckResponseMessage.isLock();
		} catch (TimeoutException toe)
		{
			throw new GlobalTransactionException(TransactionExceptionCode.io, "RPC Timeout", toe);
		} catch (RuntimeException rex)
		{
			if (Constants.developMode)
			{
				rex.printStackTrace();
			}
			log.error(rex.getMessage());
			throw new GlobalTransactionException(TransactionExceptionCode.lockable_check_failed, "Runtime", rex);
		}

	}

	//
	@Override
	public void registerResource(Resource resource)
	{
		DataSourceProxy dataSourceProxy = (DataSourceProxy) resource;
		resourceIdToResourceMap.put(dataSourceProxy.getResourceId(), dataSourceProxy);
		ResourceManagerRpcClient.getInstance().registerResource(dataSourceProxy.getResourceGroupId(), dataSourceProxy.getResourceId());
	}

	//
	public void setAsyncWorker(ResourceManagerInbound asyncWorker)
	{
		this.asyncWorker = asyncWorker;
	}

	@Override
	public void unregisterResource(Resource resource)
	{
		throw new UnsupportedOperationException("unregister a resource");
	}

}
